package com.cet.spring_batch_study.jobs.dbReader;

import com.cet.spring_batch_study.jobs.dbReader.domain.CustomerCredit;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.database.JdbcCursorItemReader;
import org.springframework.batch.item.database.JdbcPagingItemReader;
import org.springframework.batch.item.database.JpaPagingItemReader;
import org.springframework.batch.item.database.PagingQueryProvider;
import org.springframework.batch.item.database.builder.JdbcCursorItemReaderBuilder;
import org.springframework.batch.item.database.builder.JdbcPagingItemReaderBuilder;
import org.springframework.batch.item.database.builder.JpaPagingItemReaderBuilder;
import org.springframework.batch.item.database.support.SqlPagingQueryProviderFactoryBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.DataClassRowMapper;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;
import java.util.HashMap;
import java.util.Map;
import java.util.PrimitiveIterator;

/**
 * @author NSOC
 */
@Slf4j
@Configuration
public class DatabaseReaderConfig {

    @Bean
    public Job dbDataIngestingJob(JobRepository jobRepository,
                                  Step dbDataReadAndDisplayStep) {
        return new JobBuilder("dbDataIngestingJob", jobRepository)
                .start(dbDataReadAndDisplayStep)
                .build();
    }

    @Bean
    public Step dbDataReadAndDisplayStep (JobRepository jobRepository,
                                     PlatformTransactionManager transactionManager,
                                     ItemReader<CustomerCredit> creditJdbcCursorItemReader,
                                     ItemWriter<CustomerCredit> customerCreditItemWriter) {

        return new StepBuilder("dbDataReadAndDisplayStep", jobRepository)
                .<CustomerCredit, CustomerCredit> chunk(20, transactionManager)
                .reader(creditJdbcCursorItemReader)
                .writer(customerCreditItemWriter)
                .build();
    }

    @Bean
    public JdbcCursorItemReader<CustomerCredit> creditJdbcCursorItemReader(DataSource batchDatasource) {
        return new JdbcCursorItemReaderBuilder<CustomerCredit>()
                .name("creditReader")
                .dataSource(batchDatasource)
                .sql("select id, name, credit from customer")
//                .saveState(false)
                .rowMapper(new DataClassRowMapper<>(CustomerCredit.class))
                .build();
    }

    @Bean
    public JdbcPagingItemReader<CustomerCredit> itemReader(DataSource batchDatasource, PagingQueryProvider queryProvider) {
        Map<String, Object> parameterValues = new HashMap<>();
        parameterValues.put("status", "NEW");

        return new JdbcPagingItemReaderBuilder<CustomerCredit>()
                .name("creditPagingItemReader")
                .dataSource(batchDatasource)
                .queryProvider(queryProvider)
                .parameterValues(parameterValues)
                .rowMapper(new DataClassRowMapper<>(CustomerCredit.class))
                .pageSize(1000)
                .build();
    }

    @Bean
    public SqlPagingQueryProviderFactoryBean queryProvider(DataSource batchDatasource) {
        SqlPagingQueryProviderFactoryBean provider = new SqlPagingQueryProviderFactoryBean();

//        provider.setDatabaseType("MySQL");
        provider.setDataSource(batchDatasource);

        provider.setSelectClause("select id, name, credit");
        provider.setFromClause("from customer");
//        provider.setWhereClause("where status=:status");
        provider.setSortKey("id");

        return provider;
    }

//    @Bean
//    public JpaPagingItemReader itemReader() {
//        return new JpaPagingItemReaderBuilder<CustomerCredit>()
//                .name("creditReader")
//                .entityManagerFactory(entityManagerFactory())
//                .queryString("select c from CustomerCredit c")
//                .pageSize(1000)
//                .build();
//    }
}
